[fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close - #26433

…sumer-unsubscribe

…iew findings

Addresses lhotari's review on the PR:

1. Failed metadata delete no longer wedges the group: the coordinator's
   unregisterConsumer now deletes the persisted registration first and
   only on success removes the in-memory session, cancels its grace
   timer, and rebalances. A failure changes nothing — the session stays
   registered and connected, so the channelInactive → grace fallback the
   ServerCnx error path relies on actually works, and a retried
   unsubscribe retries the deletion instead of short-circuiting on an
   already-removed session.

2. close() can no longer be raced by overlapping reconnect attempts: a
   connection drop used to schedule two reconnects (failed pending
   request + connectionClosed), and close() chained its unsubscribe on a
   single last-attempt slot that either could overwrite — letting the
   older attempt register after the unsubscribe. Reconnect scheduling is
   now collapsed to one pending attempt, and close() chains on every
   in-flight subscribe attempt.

3. close() deregisters the local session immediately and unconditionally
   again: deferring removeScalableConsumerSession behind the subscribe
   future meant an unanswered subscribe retained the closed session (and
   pinned the pooled connection non-idle) for the connection's lifetime.
   Only the wire command waits for the in-flight attempts now.

4. Both scalable session requests (subscribe and unsubscribe) go through
   a new ClientCnx.sendScalableSessionRequest helper with the standard
   bookkeeping — request-timeout tracking and write-failure completion —
   instead of hand-rolled pendingRequests entries that never timed out
   and, on write failure, left the future forever pending.

5. Direct coverage for the command: a Commands round-trip test, a
   ServerCnx test for the idempotency contract and the error path (error
   response keeps the registration ref, a retry re-runs the unregister,
   success removes the ref), and a subscribe/close churn e2e asserting
   the group converges back to a sole Exclusive owner under the default
   grace period. The rejoin e2e's leaver is now tracked for cleanup so an
   early assertion failure cannot leak it onto the shared client.

6. ScalableTopicService.unregisterConsumer javadoc: the no-op branch is
   reached by deleted topics / shutdown / never-registered consumers —
   not by moved leadership, which keeps its controller entry and fails
   via checkLeader() into the ref-preserving error path.

Assisted-by: Claude Code (Fable 5)

…ations

Two follow-up findings from review:

1. The unsubscribe could overtake a registration still in flight: the
   per-connection ref was recorded only after registerConsumer resolved,
   and since the subscribe now carries a request timeout, the client can
   give up (and close) while the broker-side registration is still
   running — the unsubscribe then found no ref, answered idempotent
   success, and the registration completed afterwards as a zombie. The
   ref is now recorded before the registration resolves and carries the
   registration future; the unsubscribe handler and the channelInactive
   sweep both chain behind it, so ordering no longer depends on how long
   the client was able to wait. A failed registration drops the ref.
   Both consumer builders also close the session when the subscribe
   fails, so an abandoned attempt still sends its unsubscribe once the
   in-flight registration settles.

2. The delete-first unregister left one window on the success path: a
   same-name rejoin during the in-flight delete takes the reconnect
   branch (attaching a new consumer id), and the delete's continuation
   then removed that very session and left the persisted entry erased.
   The removal is now identity-guarded by the leaving consumer's id
   (plumbed down from the connection's registration map): a stale-id
   leave keeps the rejoined session and restores the persisted
   registration the delete erased.

Tests: a ServerCnx test pinning the ordering (an unsubscribe sent during
an in-flight registration answers nothing until the registration
completes, then unregisters); a coordinator identity test (stale-id
leave keeps the rejoined session, live-id leave removes it); and a
same-name close/resubscribe churn leg in the e2e.

Assisted-by: Claude Code (Fable 5)